﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using MessageServer.channels;
using MessageServer.config;
using MessageServer.pool;
using System.Threading;
using MessageServer.serialize;
using MessageServer.utilities;
using MessageServer.Messages;

namespace MessageServer.persistence
{
    class MessagePersistenceHandler : UpStreamHandler, DownStreamHandler
    {
        MessageServerConfig _config;
        TuningParams _params;

        public MessagePersistenceHandler(MessageServerConfig config, TuningParams parms)
        {
            _config = config;
            _params = parms;
        }

        public void Dispose()
        {
        }

        //the name by which this handler will be recognized
        public String handle { get { return "MessagePersister"; } }

        public HandlerState handleUpStream(HandlerContext ctx, ChannelEvent evt)
        {
            if (evt is UpstreamMessageEvent)
            {
                PersistentContext context;
                context = new PersistentContext(_config);
                UpstreamMessageEvent msg = evt as UpstreamMessageEvent;
                List<Object> msgs = (List<Object>)msg.getMessage();
                //PooledPersister persister = PooledPersister.getInstance(_config, _params);
                //int persisted = persister.persistObjects(msgs, evt.channel.remoteaddress.ToString());
                foreach (Object onemsg in msgs)
                {
                    int persisted = 0;
                    if (onemsg.GetType().IsDefined(typeof(MsgPersistable), false))
                    {
                        ReceivedMessage test = new ReceivedMessage(context, onemsg);
                        test.retrieve();
                        if (test.status == messagestatus.invalid)
                        {
                            ReceivedMessage rmsg = new ReceivedMessage(context, onemsg, evt.channel.remoteaddress.ToString());
                            persisted = rmsg.persist();
                        }
                    }
                    else
                        persisted = 1;

                    if (persisted > 0)
                    {
                        //TODO: Send the scknowledgement here. Will add it once the format 
                        //is defined.

                    }
                    else
                    {
                        //we have obviously received this because ack was not sent previously
                        //we will have to somehow ensure that the message is processed again to
                        //send responses here.
                    }
                }
            }
            else if (evt is ChannelStateEvent)
            {
                //handle the sent state event to update the sent status
                ChannelStateEvent state = evt as ChannelStateEvent;
                switch (state.state)
                {
                    case ChannelStateEvent.ChannelState.SENT:
                        {
                            ByteBuffer msg = state.value as ByteBuffer;
                            Object sentmsg = msg.associated;
                            PooledPersister persister = PooledPersister.getInstance(_config, _params);
                            persister.markSent(sentmsg);
                        }
                        break;
                    default:
                        break;
                }
            }

            evt.tracker.result = HandlerState.gonext;
            ctx.streamUp(evt);
            return HandlerState.gonext;
        }

        public HandlerState handleDownStream(HandlerContext ctx, ChannelEvent evt)
        {
            if (evt is DownStreamMsgEvent)
            {
                DownStreamMsgEvent dmsg = evt as DownStreamMsgEvent;
                Object msg = dmsg.getMessage();
                Object replyto = dmsg.inReplyTo;

                PooledPersister persister = PooledPersister.getInstance(_config, _params);
                int persisted = persister.persistResponse(msg, replyto, evt.channel.remoteaddress.ToString());
                if (persisted > 0)
                {
                    //TODO: Send the scknowledgement here. Will add it once the format 
                    //is defined.

                }
                else
                {
                    //we have obviously received this because ack was not sent previously
                    //we will have to somehow ensure that the message is processed again to
                    //send responses here.
                }
            }

            evt.tracker.result = HandlerState.gonext;
            ctx.streamDown(evt);
            return HandlerState.gonext;
        }
    }
}
